package com.cl.ks.flow.util;

import com.alibaba.fastjson.JSONObject;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.factory.SparkNodeFactory;
import com.cl.ks.entity.KsRule;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.functions;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Arrays;
import java.util.Collections;

@Component
public class RuleUtil {

    private final SparkNodeFactory sparkNodeFactory;

    public RuleUtil(SparkNodeFactory sparkNodeFactory) {
        this.sparkNodeFactory = sparkNodeFactory;
    }

    public SparkResult handleRule(KsRule ksRule, SparkResult originalSparkResult) {
        JSONObject ruleExpression = new JSONObject();
        SparkParam sparkParam = new SparkParam();
        ruleExpression.put("ruleExpression", ksRule.getRuleExpressionJson());
        sparkParam.setNodeExpression(ruleExpression);
        sparkParam.setSparkResultList(Collections.singletonList(originalSparkResult));
        SparkResult sparkResult = sparkNodeFactory.getSparkNodeByCode("/rule/handle").handle(sparkParam);
        Dataset<Row> res = sparkResult.getDataset();
        res = res.withColumn("ruleId", functions.lit(ksRule.getId()))
                .withColumn("ruleName", functions.lit(ksRule.getRuleName()));
        sparkResult.setDataset(res);
        return sparkResult;
    }

    public static String powerConvert(String power, Row item) {
        if (Arrays.asList(item.schema().fieldNames()).contains("rybm")) {
            String rybm = item.getAs("rybm");
            if (StringUtils.hasLength(rybm) && !rybm.equals("null")) {
                return power + ",rybm_" + rybm + ",";
            }
        }
        return power;


    }
//    public SparkResult handleRule(String ruleId,String ruleName,String ruleExpressionStr, SparkResult originalSparkResult) {
//        JSONObject ruleExpression = new JSONObject();
//        SparkParam sparkParam = new SparkParam();
//        ruleExpression.put("ruleExpression", ruleExpressionStr);
//        sparkParam.setNodeExpression(ruleExpression);
//        sparkParam.setSparkResultList(Collections.singletonList(originalSparkResult));
//        SparkResult sparkResult = sparkNodeFactory.getSparkNodeByCode("/rule/handle").process(sparkParam);
//        Dataset<Row> res = sparkResult.getDataset();
//        res = res.withColumn("ruleId", functions.lit(ruleId))
//                .withColumn("ruleName", functions.lit(ruleName));
//        sparkResult.setDataset(res);
//        return sparkResult;
//    }
}
